#include "Logger.h"
#include "EventLoop.h"
#include <cassert>
#include <cerrno>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <iostream>
#include <mutex>
#include <sstream>
#include <sys/eventfd.h>
#include <unistd.h>
#include <utility>

thread_local EventLoop* t_loopInThisThread = 0;
int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if(evtfd < 0)
    {
        Logger::Error("createEventfd");
        abort();
    }
    return evtfd;
}

EventLoop* EventLoop::getEventLoopOfCurrentThread()
{
    return t_loopInThisThread;
}

EventLoop::EventLoop()
: _looping(false)
, _quit(false)
, _eventHandling(false)
, _callingPendingFunctors(false)
, _iteration(0)
, _tid(CurrentThread::tid())
, _poller(Poller::newDefaultPoller(this))
, _timerQueue(new TimerQueue(this))
, _wakeupFd(createEventfd())
,  _wakeupChannel(new Channel(this, _wakeupFd))
, _curActiveChannel(NULL)
{
    Logger::Debug("create EventLoop {} in thread {}", this, _tid);
    if(t_loopInThisThread)
    {
        Logger::Critical("Another EventLoop {} exists in this thread {}", t_loopInThisThread, _tid);
    }
    else
    {
        t_loopInThisThread = this;
    }
    _wakeupChannel->setReadCallback(std::bind(&EventLoop::handleRead, this));
    _wakeupChannel->enableReading();
}

EventLoop::~EventLoop()
{
    Logger::Debug("EventLoop {} of thread {} destruct in thread {}", this, _tid, CurrentThread::tid());
    _wakeupChannel->disableAll();
    _wakeupChannel->remove();
    ::close(_wakeupFd);
    t_loopInThisThread = NULL;
}

void EventLoop::loop() 
{
    assert(!_looping);
    assertInLoopThread();
    _looping = true;
    _quit = false;
    Logger::Trace("EventLoop {} start looping", this);

    while(!_quit)
    {
        _activeChannels.clear();
        _pollReturnTime = _poller->poll(10000, &_activeChannels);
        ++_iteration;
        printActiveChannels(); // debug
        _eventHandling = true;
        for(Channel* channel : _activeChannels)
        {
            _curActiveChannel = channel;
            _curActiveChannel->handleEvent(_pollReturnTime);
        }

        _curActiveChannel = NULL;
        _eventHandling = false;
        doPendingFunctors();
    }
    Logger::Trace("EventLoop {} stop looping", this);
    _looping = false;
}

void EventLoop::quit()
{
    _quit = true;
    if(!isInLoopThread())
    {
        wakeup();
    }
}

void EventLoop::runInLoop(Functor cb)
{
    if(isInLoopThread())
    {
        cb();
    }
    else
    {
        queueInLoop(std::move(cb));
    }
}

void EventLoop::queueInLoop(Functor cb)
{
    {
        std::lock_guard<mutex> lg(mutex);
        _pendingFunctors.push_back(std::move(cb));
    }
    if(!isInLoopThread() || _callingPendingFunctors)
    {
        wakeup();
    }
}

size_t EventLoop::queueSize() const
{
    std::lock_guard<mutex> lg(_mutex);
    return _pendingFunctors.size();
}

void EventLoop::updateChannel(Channel* channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    _poller->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    if(_eventHandling)
    {
        assert(_curActiveChannel == channel || 
               std::find(_activeChannels.begin(), _activeChannels.end(), channel) == _activeChannels.end());
    }
    _poller->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    return _poller->hasChannel(channel);
}

void EventLoop::abortNotInLoopThread()
{
    std::stringstream ss;
    ss << "EventLoop::abortNotInLoopThread - EventLoop " << this
        << " was created in threadId_ = " << _tid
        << ", current thread id = " <<  CurrentThread::tid();
    Logger::Critical(ss.str().c_str());
}

void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = ::write(_wakeupFd, &one, sizeof one);
    if(n != sizeof one)
    {
        Logger::Error("EventLoop::wakeup() writes {} bytes instead of 8", n);
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = ::read(_wakeupFd, &one, sizeof one);
    if(n != sizeof one)
    {
        Logger::Error("EventLoop::wakeup() reads {} bytes instead of 8", n);
    }
}

void EventLoop::doPendingFunctors()
{
    _callingPendingFunctors = true;
    vector<Functor> functors;
    {
        std::lock_guard<mutex> lg(_mutex);
        functors.swap(_pendingFunctors);
    }
    for(auto &functor : functors)
    {
        functor();
    }
    _callingPendingFunctors = false;
}

void EventLoop::printActiveChannels() const
{
    for (const Channel* channel : _activeChannels)
    {
        Logger::Trace("{}", channel->reventsToString());
    }
}

